package org.sbear.dubbo.filter;

import lombok.Data;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * tpmonitor过滤器
 *
 * @author xxyWi
 * @date 2021/01/27
 */
@Activate(group = {CommonConstants.CONSUMER, CommonConstants.PROVIDER})
public class TPMonitorFilter implements org.apache.dubbo.rpc.Filter {

    @Data
    class InvokeRecord {
        private long startTime;
        private long consumingTime;

        public InvokeRecord(long startTime, long consumingTime) {
            setStartTime(startTime);
            setConsumingTime(consumingTime);
        }
    }

    private final Map<String, LinkedList<InvokeRecord>> methodCache = new ConcurrentHashMap<>();

    public TPMonitorFilter() {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            clearExpireExport();
            monitorPrintTP();
        }, 0, 5, TimeUnit.SECONDS);
    }

    private void catchRecord(Invocation invocation, long startTime) {
        synchronized (methodCache) {
            long endTime = System.currentTimeMillis();
            //没有缓存就创建新的列表
            String methodName = invocation.getMethodName();
            if (!methodCache.containsKey(methodName)) {
                methodCache.put(methodName, new LinkedList<InvokeRecord>());
            }
            //拿到链表,放入本次的
            List<InvokeRecord> invokeRecords = methodCache.get(methodName);
            invokeRecords.add(new InvokeRecord(startTime, endTime - startTime));
        }
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        Result result = null;
        long startTime = System.currentTimeMillis();
        try {
            result = invoker.invoke(invocation);
            if (result.getException() instanceof Exception) {
                throw new Exception(result.getException());
            }
        } catch (Exception e) {
            System.out.println("IP:" + RpcContext.getContext().getRemoteHost()
                    + ", Service:" + invoker.getInterface().getName()
                    + ", Method:" + invocation.getMethodName()
                    + ", Exception:{" + e.toString() + "}.");
        } finally {
            System.out.println("IP:" + RpcContext.getContext().getRemoteHost()
                    + ", Service:" + invoker.getInterface().getName()
                    + ", Method:" + invocation.getMethodName()
                    + ", Consumer:{" + (System.currentTimeMillis() - startTime) + "}.");
            catchRecord(invocation, startTime);
        }
        return result;
    }

    /**
     * 打印监控信息
     */
    private void monitorPrintTP() {
        System.out.println("打印监控【" + new Date());
        for (Map.Entry<String, LinkedList<InvokeRecord>> entryList : methodCache.entrySet()) {
            LinkedList<InvokeRecord> records = entryList.getValue();
            if (records == null || records.size() == 0) {
                continue;
            }
            LinkedList<InvokeRecord> copyRecords = new LinkedList<>(records);
            copyRecords.sort(Comparator.comparing(InvokeRecord::getConsumingTime));
            //计算TP90的index
            int size = copyRecords.size();
            int tp90Index = (int) (size * 0.9);
            int tp99Index = (int) (size * 0.99);
            System.out.println(entryList.getKey()
                    + "====>>>> TP90: " + copyRecords.get(tp90Index).getConsumingTime()
                    + ";TP99: " + copyRecords.get(tp99Index).getConsumingTime());
        }
        System.out.println("打印完毕】");
        System.out.println();
        System.out.println();
    }

    /**
     * 清空一分钟前的数据
     */
    private void clearExpireExport() {

        synchronized (methodCache) {
            //只统计1分钟，1分钟前的就过期了
            long expireTime = System.currentTimeMillis() - 1 * 60 * 1000;
            for (Map.Entry<String, LinkedList<InvokeRecord>> entryList : methodCache.entrySet()) {
                LinkedList<InvokeRecord> records = entryList.getValue();
                while (records.size() > 0) {
                    if (records.getFirst().getStartTime() < expireTime) {
                        records.removeFirst();
                    } else {
                        break;
                    }
                }
            }
        }
    }
}
